Fix streaming run key colliding across sequential runs - #13811

A run driven straight through `process_api`, without going through the
queue, has no event id, and the previous fallback minted a fresh uuid
on every chunk. Over HTTP that cannot happen, because
`restore_session_state` hands back no iterator without an event id, but
a direct caller can resume a run and would then get a new playlist URL
per chunk.

Key those runs off the iterator instead, through a `WeakKeyDictionary`
so the entry goes away with the iterator and a reused address never
becomes a second run's key.

An event id identifies an event, not a generator run, and one event
can drive two runs. Cancelling a streaming event deletes its entry
from app.iterators and adds the id to app.iterators_to_reset, so the
next call for that event gets iterator=None and process_api starts the
generator over under the same id. The restarted run then inherits the
streams the cancel had already ended, and its playlist URL does not
change, which leaves the player on a source it is done with.

The iterator is the only object that is one-to-one with a run, so key
the run on that. id() is not usable because it only holds for objects
that are alive at the same time, which is what #13809 is about, so
hold a uuid4 against the iterator in a WeakKeyDictionary instead. The
entry dies with the iterator, so a later iterator that lands on the
same address gets a fresh key.

swap_blocks carries pending_streams and pending_diff_streams across a
hot reload; it now carries the run keys that index them too. The
exception path in call_process_api goes back through app.iterators to
find the run, as it did before the event id was used.

Two problems the third self-review round found.

handle_streaming_outputs files an entry under the run key for every
generator, streaming outputs or not, and only drops it when the
session disconnects. Reused keys used to cap that by accident, so
making keys unique turned a self-limiting leak into a linear one:
500 sequential runs of a plain text generator left 165 entries
before and 500 after. A run that ends without having opened a
stream now drops its entry, which nothing could fetch anyway, and
the same 500 runs leave none. Entries holding a stream still stay,
since the playlist is fetched after the run ends.

The TypeError fallback minted a fresh uuid per call, so an iterator
that takes no weak reference got a different key on every chunk and
its playback restarted each time. The iterator is alive for the
whole run, so its address is stable there; reuse across runs is
still possible but that is what main already does.

The test that named the collision could not fail: it asserted 200
uuid4 hexes were distinct, which they are by construction, and it
passed just as well with a strong dict. Replace it with one that
drives process_api and counts run keys, which fails 80 == 100 on
main, and one that asserts the keys are released with their
iterators. Both run without ffmpeg, so the suite keeps real
coverage on a machine that has none.

The cancel test's end_stream() loop was dead, since driving the
generator to exhaustion already ends every stream. Drop it and name
the test for what it checks: two runs under one event id.

The TypeError fallback contradicted the docstring right above it: it
promised the key would never repeat and then returned str(id()),
which is the repeat this whole change is about. Nothing reachable
triggers the branch, since call_function only ever produces a
SyncToAsyncIterator or a real async generator and both take weak
references, so the only question it answers is what happens if that
stops being true. A stack trace at this line beats silently going
back to address-keyed runs.

Removing it is also what makes the exception path in
call_process_api safe to turn into a plain lookup. Minting a key
there wrote to _stream_run_ids while another exception was in
flight, to find streams that by definition do not exist, and ran
user __hash__ that could replace the exception being handled. With
no fallback left, any iterator that reaches the lookup has already
succeeded as a key on an earlier chunk, so reading it cannot fail
in a way the happy path would not have failed first.

Also pop the empty entry instead of deleting it, since the proof
that the key is present rests on there being no await in between,
and assert the run keys are distinct rather than counting retained
entries, which pinned a retention policy the test does not own. The
segment and playlist-file routes now have coverage too: all three
422 on every uuid key if run goes back to int, and only one of them
was tested.

A cancelled run never reaches its final chunk, so nothing ends its
MediaStreams and the player goes on polling a playlist that never
gets its #EXT-X-ENDLIST. Its pending_diff_streams entry, the last
full postprocessed output, is stranded too, and that is the one
stream dict the disconnect handler does not clear, so it stays for
the life of the process.

/cancel is the last place that still holds the iterator, and with
it the run key, so do both there: 300 cancelled runs leave 300
entries before and none after. This is not a regression from the
key change, since app.iterators keeps the iterator alive across a
cancel and no address was being reused there either. It is just
not fixable without something that identifies the run.

The lookup that needs is also the one call_process_api's handler
was making by reaching into the table directly, so give it a name.
Two callers want to find a run without minting one for an iterator
that has none, which is what a get-or-create helper does if a
caller forgets. The comment there claimed any iterator reaching the
lookup had already succeeded as a key; a call with an event id and
no session hash stores one that never opened a run, so say that
instead.

The route test asserted only "not 422", which a 500 would satisfy;
both routes 404 on a missing entry, so assert that.

The cleanup added to /cancel cannot fire. The route awaits
cancel_tasks first, which gathers the cancelled tasks to
completion, and their finally calls Queue.reset_iterators, which
deletes app.iterators[event_id]. By the time the route resumes the
event id is gone and the block is skipped.

Worse, it broke /cancel for events that had already finished.
call_process_api assigns app.iterators[event_id] after every call,
so the last chunk leaves None under a key that is never removed. A
cancel for such an event reached the new lookup, and
WeakKeyDictionary.get(None) raises rather than returning None, so
the request went from 200 to 500. gradio_client's Job.cancel posts
there unconditionally once a job is cancellable.

Ending the streams was redundant in any case: call_process_api's
exception path already does it, which is why the streams looked
closed. The test missed all of this because it drove process_api
directly and filled app.iterators by hand, so no task existed for
cancel_tasks to match and the block ran in a state the server
never reaches.

The diff-state leak the cleanup was aiming at is not a regression
from this PR, and closing it properly means covering the error
path and sibling runs too, so it belongs elsewhere. Keep the
lookup accessor, since the exception path in call_process_api
still wants a lookup that does not mint, but underscore it: Blocks
is public API and this is not.

A run that is cancelled or raises never reaches the final chunk that
would delete its pending_diff_streams entry, and that entry holds the
last full postprocessed output of every component: a whole Chatbot
history for a chat app. Colliding keys used to reclaim it, because a
later run inherited the entry and deleted it on its own final chunk,
so 300 aborted runs left 155 entries. Distinct keys leave 300, and
pending_diff_streams is the one stream dict the disconnect handler
never clears, so they outlive the session.

This is the leak the backed-out commit was aiming at, at a place that
actually runs. call_process_api's handler already ends the run's
streams on the way out and has the session hash and the run key in
hand, so it drops the diff entry there too. The same 300 runs now
leave none. A first-chunk abort still slips through, since the
iterator is not in app.iterators yet, but that is unchanged from
before.

Also capture the session's dict once in handle_streaming_outputs.
Reaching back through the defaultdict after the awaits in the output
loop can recreate a session that the disconnect handler removed in
between, leaving an entry nothing will ever clear.

Two test gaps behind this. Nothing covered the abort path, so add a
test that drives call_process_api into it. And drive_streaming_run
only recorded the first chunk's URL, so a key that churned per chunk
- the regression bcdd127 fixed - was caught only by an ffmpeg test;
it now checks the final chunk's URL against the first.

handle_streaming_outputs files an entry under the run key for every
generator, before it knows whether any of its outputs stream. The
final chunk drops it when the run never opened one, but a run that
raises or is cancelled never reaches that chunk, so the entry stayed
until the session disconnected.

Reused keys used to cap this the way they capped the finished-run
case: 200 aborted runs of a text generator leave 200 entries under
distinct keys, against 154 to 157 across repeat runs under the old
id(iterator) key. The exception path already ends the run's streams
and drops its diff state, so drop the stream entry there too when it
holds no stream, which takes the same 200 runs back to none.

A real cancel reaches that handler as well, before the cancelled
task's finally gets to Queue.reset_iterators, so it is covered by the
same line: one entry before, none after. Entries that do hold a
stream still stay, since the playlist is fetched after the run ends.

Found by abidlabs in review.

The comment on test_two_runs_under_one_event_id_get_separate_streams
gave the same cause for a cancel restart that was already taken out of
the PR description as inaccurate. The only writer of
app.iterators_to_reset sits inside `if body.event_id in app.iterators`,
which Queue.reset_iterators has already emptied from the cancelled
task's own finally by the time /cancel resumes, so that block does not
run. What restarts the generator is the delete itself, after which
restore_session_state hands back iterator=None. The test's conclusion
is unaffected.

The exception handler reached through blocks.pending_streams twice.
Capture the session's runs once, the way handle_streaming_outputs
already does. Drop a dead assignment in the sibling test while there.

Both abort tests looped five times over a generator that raises on its
second chunk, so the third pass was unreachable and the five read as if
the count mattered. They also carried the same driver twice. One
module-level helper, bounded at the two chunks the run actually has,
covers both.

The exception handler asked the app for its blocks twice. Bind them
once, above the lookup that wants them first.

The comment beside the diff-state pop said this handler was the only
place that dropped it and that nothing cleared the dict on disconnect.
Both stopped being true one commit ago, when the disconnect handler
started dropping it. Say what is left that is still true, which is that
the final chunk never arrives, and say it once for the two pops instead
of once each.

That handler had grown twelve comment lines around thirteen of code,
against about five percent for the file as a whole. It is down to six.
The line restating that the loop below it closes open streams goes with
them.

#13806 landed a playlist test that files its stream under an int run key
and fetches it back through the route. On this branch the route's `run`
is a `str`, so the path segment no longer parses to the int the dict is
keyed by and the fetch 404s. The key in the test is a string now. What
it asserts, that the playlist carries a stable target duration, is
untouched.

4092317 dropped a dead session's diff state from the heartbeat
disconnect handler. That handler only runs for an app that opens a
heartbeat, which takes gr.State, an unload or stream listener or a
per-session cache. The issue's app, a button driving a streaming
gr.Audio, has none of those, so for the app this PR is about the
cleanup never ran.

The place every queued event does pass through is the finally in
Queue.process_events, whether the run finished, raised, was cancelled
or lost its client, and it already resets the event's iterator there.
Close out the run's streams first, while the iterator that keys them is
still stored, through one Blocks helper that the exception path in
call_process_api now calls as well. The heartbeat handler goes back to
what it was.

The number that justified 4092317 was also wrong. It came from a loop
that popped app.iterators by hand with nothing allocating in between,
so id() reuse reclaimed half the entries. Driven through a real
connection, with a queue join and an SSE stream between runs, 100
interrupted runs leave 98 entries under the old key and 100 under this
PR's, so the leak is pre-existing and all but unchanged by the key
switch. It is closed here anyway, since the entry holds the last full
output of every component for the life of the process. The same 100
runs now leave none.

The disconnect test drives a launched app over a real connection and
fails without the queueing.py change.

handle_streaming_outputs filed pending_streams[session][run] for every
generator up front and two other places removed it again when it turned
out to hold nothing: the final chunk, and _drop_run_streams for a run
that never got there. File it where a MediaStream is first constructed
instead. A run with no streaming output now never touches the dict, so
there is nothing to remove on any path, and the two removals go.

A run whose client went away kept its streams, on the rule that a
playlist is fetched after the run ends. That rule is for a client that
is still there. event.alive is False only after clean_events, which
only the /queue/data disconnect reaches, so the queue's finally now
passes orphaned=not event.alive and the helper drops the run's entry
outright in that case. Without it an abandoned audio run on an app with
no heartbeat pinned its segments for the life of the process, since the
heartbeat handler never runs for such an app.

The heartbeat handler now drops a session's pending_diff_streams beside
pending_streams. A generator driven through /run/{api_name} with a
session hash never enters the queue, so nothing else reaches its diff
state when the session goes away, and the per-session dict itself was
never removed.

Two orderings hold this together, so the finally now says so. /cancel
awaits this task before dropping the iterator itself, and both cleanup
sites look the run up through app.iterators. And a finished run is
still alive at the finally, since nothing awaits between its completion
message and that point; the server, not the client, closes the SSE
stream once a session has no pending events left.

The disconnect test now streams into gr.Audio, so it covers the dropped
stream, and it checks the run is filed while the connection is still
open rather than racing the teardown after closing it. A new test drives
a streaming run to completion over a real connection and fetches its
playlist afterwards, the half of the rule the orphan drop must not
break.

Found by self-review.

6bedc27 had the heartbeat disconnect handler drop a session's
pending_diff_streams beside its streams. A heartbeat can die while the
/queue/data stream that carries a run survives, and the handler does
not stop the run, so its next chunk found no diff state, went out as a
full value, and the client applied it as a diff and failed the run.
That line goes; the queue's finally already drops a run's diff state
when the run ends.

handle_streaming_outputs re-filed the dict it had read before awaiting
stream_output. A session torn down during that await left the local
holding an already-ended stream, which setdefault then put back under
the same run key, and later chunks appended to it. Read the entry
fresh at the point of use instead.

Both dicts kept an empty per-session dict once its last run was gone,
one per visitor session for the life of the process on an app with no
heartbeat. A run's entry and a session dict it leaves empty now go
together, through one small helper used by the finally path and the
final-chunk path alike.

The finally reaches the shared dicts through the queue's own Blocks,
which cannot raise the way app.get_blocks() can during teardown, and
its comment no longer argues about finished runs: their iterator is
already None by then.

Tests: the final-chunk test asserted through a defaultdict subscript
that created the very entry it checked, and the disconnect test's
generator could finish on its own before a slow runner noticed the
disconnect. The first now checks the session key is absent; the second
gates the generator on an event the test sets on the way out.

Found by self-review.

handle_streaming_diffs filed an entry for every generator run, but its
later chunks are only fetchable under an event id: restore_session_state
returns no iterator without one. A /run/{api_name} call, which has none,
makes one call and returns, so every such call left an entry holding the
full postprocessed output for the life of the process, as did every
.stream() chunk after the first, which arrives without an id. Under the
old key a collision reclaimed one in twenty of those; under this PR's
none. process_api now passes no run to the diff bookkeeping when there
is no event id. The first chunk's output is unchanged, since the first
chunk of a run never carried a diff.

The queue's finally looked the session up on event.data, which
/stream/{event_id} replaces with the client's payload; a caller that
leaves session_hash out of it would have skipped the cleanup. Event
keeps the session hash it was created with, and push writes that same
value into the body, so the finally reads it from the event.

Also record on _stream_run_ids what its weak keys rest on, and shorten
the heartbeat handler's note to what it does rather than why.

Found by self-review.

The queue's finally checked the server app once per event, inside the
loop that also resets iterators, and that reset raises on the same None
the check guards against. The close-out now has its own small loop with
the check outside it, and the reset loop is back to what it was.

The diff bookkeeping comment now says what a call without an event id
gets, full values, since that is what /run/{api_name} callers and the
.stream() client read.

The two tests that launch a server used httpx's default five second
read timeout and gave the disconnect five seconds to be noticed. Both
now say 30 and 15 seconds, so a slow runner reads as a slow runner.

Found by self-review.